package cn.doitedu.day02

import org.apache.spark.rdd.{RDD, ShuffledRDD}
import org.apache.spark.{Aggregator, HashPartitioner, SparkConf, SparkContext}

object T14_ShuffledRDDDemo{

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf
    val conf = new SparkConf().setAppName("MapPartitionsWithIndexDemo")
      .setMaster("local[4]")

    //2.创建SparkContext
    val sc = new SparkContext(conf)

    val lst = List(
      ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
      ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
      ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
      ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
    )
    //通过并行化的方式创建RDD，分区数量为4
    val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)

    //ShuffledRDD的三个功能1.分区 2.分组或聚合 3.shuffle的同时在分区内排序
    val shuffledRDD = new ShuffledRDD[String, Int, Int](wordAndOne, new HashPartitioner(wordAndOne.partitions.length))

    //设置mapsideCombine = true的情况
    //在上游分区中，key第一次出现对应的value的计算逻辑
    val f1 = (x: Int) => x
    //在上游分区中，key再次出现对应的value的计算逻辑
    val f2 = (a: Int, b: Int) => a + b
    //在下游，key相同的value的计算逻辑
    val f3 = (m: Int, n: Int) => m + n

    val aggregator = new Aggregator[String, Int, Int](
      f1, f2, f3
    )

    //设置Shuffle的聚合器
    shuffledRDD.setAggregator(aggregator);

    //设置mapSideCombine 为 true或false
    shuffledRDD.setMapSideCombine(true)

    //触发Action
    shuffledRDD.saveAsTextFile("out/out01")

  }


}
